package com.booter.ws.config;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.HttpResponseInterceptor;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.message.BasicHeader;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.nio.reactor.IOReactorException;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PreDestroy;

@Configuration
public class EsConfiguration {

    @Value("${spring.elasticsearch.password:Lpo*owegdsEUi3#q!}")
    private String password;

    private IdleConnectionMonitorThread idleThread;

    @Bean(destroyMethod = "close")
    public ElasticsearchClient elasticsearchClient() throws IOReactorException {
        RestClientBuilder builder = RestClient.builder(new HttpHost("es-cn-y1i3s9g980009cv5e.public.elasticsearch.aliyuncs.com", 9200, "http"));
        BasicCredentialsProvider basicCredentialsProvider = new BasicCredentialsProvider();
        basicCredentialsProvider.setCredentials(
                AuthScope.ANY, new UsernamePasswordCredentials("elastic", password)
        );
        IOReactorConfig ioReactorConfig = IOReactorConfig.custom().
                setIoThreadCount(Runtime.getRuntime().availableProcessors())
                .setSoKeepAlive(true)
                .build();
        ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig); //创建一个ioReactor
        PoolingNHttpClientConnectionManager cm = new PoolingNHttpClientConnectionManager(ioReactor);
        //new PoolingNHttpClientConnectionManager(new DefaultConnectingIOReactor())
        // 设置最大连接数
        cm.setMaxTotal(20);
        // 将每个路由默认最大连接数
        cm.setDefaultMaxPerRoute(20);
        idleThread = new IdleConnectionMonitorThread(cm);
        idleThread.start();
//        HttpHost httpHost = new HttpHost(hostname, port);
        // 设置目标主机对应的路由的最大连接数，会覆盖setDefaultMaxPerRoute设置的默认值
//        cm.setMaxPerRoute(new HttpRoute(httpHost), 20);
        builder.setHttpClientConfigCallback(b -> {
            b.setDefaultCredentialsProvider(basicCredentialsProvider);
            b.setConnectionManager(cm);
            //Missing [X-Elastic-Product] header. Please check that you are connecting to an Elasticsearch instance
            b.addInterceptorLast((HttpResponseInterceptor) (response, context) -> response.addHeader("X-Elastic-Product", "Elasticsearch"));
            return b;
        });
        //设置header Content-type 否则406异常，{"error":"Content-Type header [application/vnd.elasticsearch+json; compatible-with=8] is not supported","status":406}
        Header[] headers = new Header[]{new BasicHeader("Content-type", "application/json")
//                , new BasicHeader("X-Elastic-Product", "Elasticsearch")
        };
        RestClient restClient = builder.setDefaultHeaders(headers).build();
        ElasticsearchTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
        return new ElasticsearchClient(transport);
//        ElasticsearchAsyncClient client = new ElasticsearchAsyncClient(transport);
    }


    @PreDestroy
    public void close() {
        idleThread.shutdown();
    }


}
